package com.rrd.dw.mr.gzip.bhzx;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.rrd.dw.mr.key.IntPair;
import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.HadoopUtils;
import com.rrd.dw.utils.MD5Util;
/**
 * 逻辑口径变更之前的程序，不包含垫付，都按百行正常的逾期计算
 * @author liubaoxin
 *
 */
public class BHZXD3MR extends Configured implements Tool {
	public static final String CRT_FLAG = "\001";
    public static final Pattern CRT_PATTERN = Pattern.compile(CRT_FLAG);
    
	public static class D3Maper extends Mapper<LongWritable, Text, IntPair, Text> {
		 private IntPair ekey = new IntPair();
		private String line = "";
		private String arr[]=null;
		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, IntPair, Text>.Context context)
				throws IOException, InterruptedException {
			if(value==null){
				return;
			}
			line=value.toString();
			arr=CRT_PATTERN.split(line);
			if(arr==null || arr.length<17){
				return;
			}
		
			
			ekey.set(arr[0], Integer.parseInt(arr[7]));
			context.write(ekey, value);
		}
	}
	
	public static class D3Reduce extends Reducer<IntPair, Text, NullWritable, Text> {
		private final static int K=0;
		private static NullWritable key_null = NullWritable.get();
		private Text val_Text = new Text();
		private List<String> list = new ArrayList<String>(50);
		private List<String[]> plan_list = new ArrayList<String[]>(200);
		private List<Integer> payoff_list = new ArrayList<Integer>(100);
		private Map<String,String> parseLine(String arr[] ){
			Map<String,String> map = new HashMap<String,String>(50);
			map.put("loan_key", arr[0]);
			map.put("loan_id", arr[1]);
			map.put("loan_payoff_time",arr[2] );
			map.put("real_repayment", arr[3]);
			map.put("loan_amt", arr[4]);
			map.put("loan_status", arr[5]);
			map.put("repay_plan_key", arr[6]);
			map.put("term_no", arr[7]);
			map.put("target_repayment_date", arr[8]);
			map.put("payoff_time", arr[9]);
			map.put("repay_type", arr[10]);
			map.put("principal", arr[11]);
			map.put("planned_payment", arr[12]);
			map.put("create_time", arr[13]);
			map.put("derate_amt", arr[14]);
			map.put("name", arr[15]);
			map.put("pid", arr[16]);
			map.put("mobile", arr[17]);
			map.put("od_days", arr[18]);
			map.put("period_real_repayment", arr[19]);
			map.put("next_due_date", arr[20]);  //下一期应还款日期
			map.put("last_due_date", arr[21]);  //最后一期应还日期
			map.put("last_period", arr[22]);  //最后一期期数
			map.put("repay_begin_period", arr[23]);  //本次已还的开始期数
			map.put("repay_end_period", arr[24]);  //本次已还的结束期数
			map.put("dt", arr[25]);  //观察日
			map.put("loan_platform", arr[26]);  //进件平台
			map.put("op_code", "A"); 
			return map;
		}
		@Override
		protected void reduce(IntPair key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			list.clear();
			plan_list.clear();
			payoff_list.clear();
			Map<String,String> map =null;
			String arr[]=null;
			double loan_amt=0;
			double planned_payment=0;
		    double due_loan_amount=0;
		    double act_repay_amt=0; 
			double overdue_amt=0;
			double derate_amt=0; //减免金额
			double real_repayment=0;
			double period_real_repayment=0;
			double remaining_amount=0;
			int repay_begin_period=-1;
			int repay_end_period=-1;
			int old_days=0;
			int term_no=-1;
			String min_due_date="";
			int i=0;
			list.clear();
			plan_list.clear();
			String confirm_at="";
			String uploadTs="";
			String target_repayment_date=null;
			String target_repayment_date_k=null;
			String next_target_repayment_date_k=null;
			for(Text t :values){
				arr = CRT_PATTERN.split(t.toString());
				if(arr[9].equals("\\N")){
					arr[9]="";
				}
				map=parseLine(arr);
				
				plan_list.add(arr.clone());
				loan_amt=Double.parseDouble(map.get("loan_amt"));
				planned_payment=Double.parseDouble(map.get("planned_payment"));
				real_repayment=Double.parseDouble(map.get("real_repayment")); 
				period_real_repayment=Double.parseDouble(map.get("period_real_repayment")); 
				derate_amt=Double.parseDouble(map.get("derate_amt")); 
				term_no=Integer.parseInt(map.get("term_no"));
				repay_begin_period=Integer.parseInt(map.get("repay_begin_period"));
				repay_end_period=Integer.parseInt(map.get("repay_end_period"));
				target_repayment_date=map.get("target_repayment_date").substring(0,10);
				target_repayment_date_k=DateUtil.formateDateStr(DateUtil.addDay(DateUtil.formateToDate(map.get("target_repayment_date"),"yyyy-MM-dd"), K+1),"yyyy-MM-dd");
				if(map.get("next_due_date")!=null && !map.get("next_due_date").equals("")){
					next_target_repayment_date_k=DateUtil.formateDateStr(DateUtil.addDay(DateUtil.formateToDate(map.get("next_due_date"),"yyyy-MM-dd"), K+1),"yyyy-MM-dd");
				}else{
					next_target_repayment_date_k=null;
				}
				
				uploadTs=DateUtil.getCurrentDateTime();
				
				if(map.get("payoff_time").equals("")  || map.get("payoff_time").substring(0,10).compareTo(target_repayment_date_k)>=0  ){
					confirm_at=target_repayment_date_k;
					if(min_due_date.equals("")){
						min_due_date=map.get("target_repayment_date");
					}
				}else{
					min_due_date="";
					confirm_at=map.get("payoff_time"); 
					//截止当前已经没有逾期	 
					 
				}
				//如果是提前还清的话，合并只报还的第一期
				if(!map.get("payoff_time").equals("") && real_repayment<=0
						&&  map.get("payoff_time").substring(0,10).compareTo(map.get("target_repayment_date").substring(0,10))<0){
					min_due_date="";
					act_repay_amt=act_repay_amt+planned_payment;
					payoff_list.add(term_no);
					continue;
				}
				//应还款次日上报
				//计算逾期天数 
				if(min_due_date.equals("")){
					old_days=0;
				}else{
					old_days=DateUtil.date_diff(confirm_at, min_due_date);
				}
				due_loan_amount=getLoan_due_amt(plan_list, term_no, confirm_at, period_real_repayment, planned_payment);
				
				overdue_amt=overdue_amt+(old_days>K?(period_real_repayment>0?period_real_repayment:planned_payment):0);
				//act_repay_amt 主要用来计算剩余金额，不是征信指标值
				act_repay_amt=act_repay_amt+(old_days>0?0:planned_payment);
				i++;
				map.put("req_id", MD5Util.getMd5(map.get("repay_plan_key")+uploadTs+i));
				map.put("uploadTs", uploadTs);
				map.put("term_status",old_days>K?"overdue":"normal");
				map.put("real_repayment_date",old_days>K?"":map.get("payoff_time")); //实际还款时间
				map.put("target_repayment",String.valueOf(due_loan_amount));
				map.put("real_repayment", old_days>K?"0":String.valueOf(real_repayment));	
				map.put("overdue_status", old_days>K?"D"+old_days:"");
				map.put("status_confirm_at",confirm_at);
				
				map.put("act_repay_amt", String.valueOf(act_repay_amt));
				remaining_amount=loan_amt-act_repay_amt;
				if((map.get("last_due_date").equals(map.get("target_repayment_date")) || repay_end_period==Integer.parseInt(map.get("last_period"))) &&  old_days<=K ){
					map.put("loan_status", "3");
					overdue_amt=0;
					remaining_amount=0;
					payoff_list.add(Integer.parseInt(map.get("term_no")));
				}else if(old_days>K){
					map.put("loan_status", "2");
				}else{
					map.put("loan_status", "1");
					payoff_list.add(Integer.parseInt(map.get("term_no")));
				} 
				map.put("remaining_amount", String.valueOf(remaining_amount));
				map.put("overdue_amount", String.valueOf(overdue_amt)); //当 前 逾期总额
				map.put("repay_type", old_days+","+map.get("last_due_date")+","+map.get("target_repayment_date")+","+confirm_at+","+map.get("dt")+","+map.get("real_repayment"));
				 
				val_Text.set(processToString(map));
				context.write(key_null, val_Text);
				
				//判断如果本期未逾期
				if(old_days<=0){
					due_loan_amount=0;
				}
				//判断如果是逾期已还,先暂存起来
//				if(!map.get("payoff_time").equals("") && real_repayment>0
//					    && map.get("payoff_time").substring(0,10).compareTo(target_repayment_date)>0){
//					 list.add(map.get("payoff_time")+","+real_repayment+","+planned_payment);
////					 System.out.println(map.get("payoff_time")+","+real_repayment+","+planned_payment);
//				}
				 if(old_days>K && real_repayment>0 && !map.get("payoff_time").equals("")){
					 list.add(map.get("payoff_time")+","+real_repayment+","+planned_payment+","+map.get("term_no"));
				 }
				 
//				处理逾期情况,并且是在下一期到期之前，已还
				if(list!=null && list.size()>0){
				for(String s:list){
					i++;
					arr=s.split(",");
//					System.out.println("target_repayment_date="+map.get("target_repayment_date")+"  , next_due_date="+map.get("next_due_date")+",arr[0]="+arr[0]);
					if(arr[0].substring(0,10).compareTo(map.get("target_repayment_date").substring(0,10))>=0
							&& (next_target_repayment_date_k==null || arr[0].substring(0,10).compareTo(next_target_repayment_date_k)<0  )){
						real_repayment=Double.parseDouble(arr[1]);
						confirm_at=arr[0];
						due_loan_amount=getLoan_due_amt(plan_list, Integer.parseInt(map.get("term_no")), arr[0], period_real_repayment, planned_payment);
						if(arr[0].equals(map.get("payoff_time"))){
							old_days=0;
							min_due_date="";
						}else{
							System.out.println("loan_id="+map.get("loan_id")+",payoff_time="+map.get("payoff_time")+",confirm_at="+confirm_at+",min_due_date=>"+min_due_date);
							if(min_due_date.equals("")){
								old_days=0;
							}else{
								old_days=DateUtil.date_diff(confirm_at, min_due_date);
							}
							
						}
						
						act_repay_amt=act_repay_amt+Double.parseDouble(arr[2]);
						remaining_amount=loan_amt-act_repay_amt;
						overdue_amt=overdue_amt-real_repayment;
						map.put("req_id", MD5Util.getMd5(map.get("repay_plan_key")+uploadTs+i));
						map.put("uploadTs", uploadTs);
						map.put("term_status",old_days>K?"overdue":"normal");
						map.put("real_repayment_date",confirm_at); //实际还款时间
						map.put("target_repayment",String.valueOf(due_loan_amount));
						map.put("real_repayment",String.valueOf(real_repayment));	
						map.put("overdue_status", old_days>K?"D"+old_days:"");
						map.put("status_confirm_at",confirm_at);
						map.put("act_repay_amt", String.valueOf(act_repay_amt));
						
						if(map.get("last_due_date").equals(map.get("target_repayment_date")) &&  old_days<=0){
							map.put("loan_status", "3");
							remaining_amount=0;
							overdue_amt=0;
						}else if(old_days>K){
							map.put("loan_status", "2");
						}else{
							map.put("loan_status", "1");
						} 
						map.put("remaining_amount", String.valueOf(remaining_amount));
						map.put("overdue_amount", String.valueOf(overdue_amt)); //当 前 逾期总额
						map.put("repay_type", old_days+","+map.get("last_due_date")+","+map.get("target_repayment_date")+","+confirm_at+","+map.get("dt")+","+map.get("real_repayment"));
						 
						val_Text.set(processToString(map));
						context.write(key_null, val_Text);
//						list.remove(s);
						payoff_list.add(Integer.parseInt(arr[3]));
					}
				}
				}
				//最后一期，整笔逾期30天以上，并且没有还清,每30天上报一次
				confirm_at=map.get("target_repayment_date"); 
				if(old_days>K && map.get("last_due_date").equals(map.get("target_repayment_date"))){
					
					while(true){
						String pre_confirm_at=confirm_at; 
//						System.out.println("pre_confirm_at="+pre_confirm_at+",confirm_at="+confirm_at);
						
						confirm_at=DateUtil.formateDateStr(DateUtil.addDay(DateUtil.formateToDate(pre_confirm_at,"yyyy-MM-dd"), 30));
						
						real_repayment=0;
						for(String s:list){
							arr=s.split(",");
							if(arr[0].substring(0,10).compareTo(pre_confirm_at)>=0
									&& arr[0].substring(0,10).compareTo(confirm_at)<0){
								real_repayment=Double.parseDouble(arr[1]);
								confirm_at=arr[0];
								old_days=DateUtil.date_diff(confirm_at, min_due_date);
								if(arr[0].equals(map.get("payoff_time"))){
									old_days=0;
									min_due_date="";
								}
								due_loan_amount=getLoan_due_amt(plan_list, Integer.parseInt(map.get("term_no")), confirm_at, period_real_repayment, planned_payment);
								i++;
								act_repay_amt=act_repay_amt+Double.parseDouble(arr[2]);
								remaining_amount=loan_amt-act_repay_amt;
								overdue_amt=overdue_amt-real_repayment;
								map.put("req_id", MD5Util.getMd5(map.get("repay_plan_key")+uploadTs+i));
								map.put("uploadTs", uploadTs);
								map.put("term_status",old_days>0?"overdue":"normal");
								map.put("real_repayment_date",confirm_at); //实际还款时间
								map.put("target_repayment",String.valueOf(due_loan_amount));
								map.put("real_repayment",String.valueOf(real_repayment));	
								map.put("overdue_status", old_days>0?"D"+old_days:"");
								map.put("status_confirm_at",confirm_at);
								map.put("act_repay_amt", String.valueOf(act_repay_amt));
								 
								if( old_days<=0){
									map.put("loan_status", "3");
									remaining_amount=0;
									overdue_amt=0;
								}else{
									map.put("loan_status", "2");
								} 
								map.put("remaining_amount", String.valueOf(remaining_amount));
								map.put("overdue_amount", String.valueOf(overdue_amt)); //当 前 逾期总额
								map.put("repay_type", old_days+","+map.get("last_due_date")+","+map.get("target_repayment_date")+","+confirm_at+","+map.get("dt")+","+map.get("real_repayment"));
								 
								val_Text.set(processToString(map));
								context.write(key_null, val_Text);
//								list.remove(s);
								payoff_list.add(Integer.parseInt(arr[3]));
							}}
						i++;
						confirm_at=DateUtil.formateDateStr(DateUtil.addDay(DateUtil.formateToDate(pre_confirm_at,"yyyy-MM-dd"), 30));
						if(confirm_at.compareTo(map.get("dt"))>0){
							break;
						}
						old_days=DateUtil.date_diff(confirm_at, min_due_date);
						due_loan_amount=getLoan_due_amt(plan_list, Integer.parseInt(map.get("term_no")), confirm_at, period_real_repayment, planned_payment);
						map.put("req_id", MD5Util.getMd5(map.get("repay_plan_key")+uploadTs+i));
						map.put("uploadTs", uploadTs);
						map.put("term_status","overdue");
						map.put("real_repayment_date",""); //实际还款时间
						map.put("target_repayment",String.valueOf(due_loan_amount));
						map.put("real_repayment", "0");	
						map.put("overdue_status", "D"+old_days);
						map.put("status_confirm_at",confirm_at);
						map.put("overdue_amount", String.valueOf(overdue_amt)); //当 前 逾期总额
						map.put("remaining_amount", String.valueOf(loan_amt-act_repay_amt));
						map.put("loan_status", "2");
						map.put("act_repay_amt", String.valueOf(act_repay_amt));
						map.put("repay_type", old_days+","+map.get("last_due_date")+","+map.get("target_repayment_date")+","+confirm_at+","+map.get("dt"));
						val_Text.set(processToString(map));
						context.write(key_null, val_Text);
						
						
					}
				}
			}
			 
		
		}
		//计算直接当期剩余应还款金额
		private double getLoan_due_amt(List<String[]> values,int term_no,String confim_date,double period_real_repayment,double paln_amount){
			double loan_due_amt=0;
			confim_date=confim_date.substring(0,10);
			String payoff_date="";
			for(String[] arr :values){
				//arr[8] target_repayment_date
				//arr[12] planned_payment
				//arr[9]  payoff_time
				//arr[19]  period_real_repayment  
				//arr[7]  term_no
				 
				payoff_date=(arr[9]==null || arr[9].equals(""))?"": arr[9].substring(0, 10);
				if(Integer.parseInt(arr[7])<=term_no && !payoff_list.contains(Integer.parseInt(arr[7]))){
					if(payoff_date.equals(confim_date.substring(0,10))){
						loan_due_amt=loan_due_amt+(Double.parseDouble(arr[19])>0?Double.parseDouble(arr[19]):Double.parseDouble(arr[12]));
					}else{
						loan_due_amt=loan_due_amt+Double.parseDouble(arr[12]);
					}
					
					 
				}
				 
             }
			
			return loan_due_amt;
		}
		
	}
	private static String processToString(Map<String,String> map){
		StringBuilder sb = new StringBuilder();
//		sb.append(MD5Util.getMd5(map.get("repay_plan_key")+map.get("status_confirm_at"))).append(CRT_FLAG);
		sb.append(map.get("loan_key")).append(CRT_FLAG);
		sb.append(map.get("repay_plan_key")).append(CRT_FLAG);
		sb.append(map.get("req_id")).append(CRT_FLAG);
		sb.append("A").append(CRT_FLAG); //op_code
		sb.append(map.get("uploadTs")).append(CRT_FLAG);
		sb.append(map.get("loan_id")).append(CRT_FLAG);
		sb.append(map.get("name")).append(CRT_FLAG);
		sb.append(map.get("pid")).append(CRT_FLAG);
		sb.append(map.get("mobile")).append(CRT_FLAG);
		sb.append(map.get("term_no")).append(CRT_FLAG);
		sb.append(map.get("term_status")).append(CRT_FLAG);
		sb.append(map.get("target_repayment_date")).append(CRT_FLAG);
		sb.append(map.get("real_repayment_date")).append(CRT_FLAG);
		sb.append(map.get("planned_payment")).append(CRT_FLAG);
		sb.append(map.get("target_repayment")).append(CRT_FLAG);
		sb.append(map.get("real_repayment")).append(CRT_FLAG);
		sb.append(map.get("overdue_status")).append(CRT_FLAG);
		sb.append(map.get("status_confirm_at")).append(CRT_FLAG);
		sb.append(map.get("overdue_amount")).append(CRT_FLAG);
		sb.append(map.get("remaining_amount")).append(CRT_FLAG);
		sb.append(map.get("loan_status")).append(CRT_FLAG);
		sb.append(map.get("act_repay_amt")).append(CRT_FLAG);
		sb.append(map.get("od_days")).append(CRT_FLAG);
		sb.append(map.get("payoff_time")).append(CRT_FLAG);
		sb.append(map.get("last_due_date")).append(CRT_FLAG);
		sb.append(map.get("loan_amt")).append(CRT_FLAG);
		sb.append(map.get("loan_payoff_time")).append(CRT_FLAG);
		sb.append(map.get("loan_platform")).append(CRT_FLAG);
		sb.append(map.get("loan_key")).append(CRT_FLAG);
		sb.append(map.get("repay_type"));
		return sb.toString();
		 
	}
	/*
     * 自定义分区函数类FirstPartitioner，根据 IntPair中的first实现分区
     */
    public static class FirstPartitioner extends Partitioner<IntPair, Text>{
        @Override
        public int getPartition(IntPair key, Text value,int numPartitions){ 
            return Math.abs(key.getFirst().hashCode() * 127) % numPartitions;
        }
    }
    
    /*
     * 自定义GroupingComparator类，实现分区内的数据分组
     */
    @SuppressWarnings("rawtypes")
    public static class GroupingComparator extends WritableComparator{
        protected GroupingComparator(){
            super(IntPair.class, true);
        }
        
        @Override
        public int compare(WritableComparable w1, WritableComparable w2){
            IntPair ip1 = (IntPair) w1;
            IntPair ip2 = (IntPair) w2;
            String l = ip1.getFirst();
            String r = ip2.getFirst();
            return l.equals(r) ? 0 : (l.compareTo(r)<0? -1 : 1);
        }
    }
	@Override
	public int run(String[] args) throws Exception {

		 Configuration conf = getConf();
	        Job job = Job.getInstance( conf);
	        conf.set("mapreduce.output.fileoutputformat.compress", "true");
	        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
	        conf.set("mapreduce.output.fileoutputformat.compress.codec",
	            "org.apache.hadoop.io.compress.GzipCodec");
	        conf.set("io.compression.codecs",
	            "org.apache.hadoop.io.compress.GzipCodec");
	         
	        
	        String inputDir = conf.get("param.input.path");
	         String outDir = conf.get("param.out.path");
	         if(inputDir==null || inputDir.equals("")){
	        	 System.out.println("请输入输入路径：param.input.path");
	        	 System.exit(0);
	         }
	         if(outDir==null || outDir.equals("")){
	        	 System.out.println("请输入输出路径：param.out.path");
	        	 System.exit(0);
	         }
	        System.out.println("param.input.path=" + inputDir);
	        System.out.println("param.out.path=" + outDir);
	        Path outpath = new Path(outDir);
	        HadoopUtils.delete(conf, outpath);  
	        TextInputFormat.setInputPaths(job, new Path(inputDir));
	        TextOutputFormat.setOutputPath(job, new Path(outDir));
	        FileOutputFormat.setCompressOutput(job, true);
	        FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); //设置压缩格式
	        job.setMapperClass(D3Maper.class);
	        job.setReducerClass(D3Reduce.class);
	        job.setPartitionerClass(FirstPartitioner.class);
	        job.setGroupingComparatorClass(GroupingComparator.class);
	        job.setOutputKeyClass(Text.class);
	        job.setMapOutputKeyClass(IntPair.class);
	        job.setNumReduceTasks(5);
	        job.setJarByClass(this.getClass());
	        
 
	        return job.waitForCompletion(true) ? 0 : 1;
		 
	}
	public static void main(String[] args) throws Exception {
		 int res = ToolRunner.run(new BHZXD3MR(), args);
	        System.exit(res);
		
	}
	
	

}

